package prictice.transaction1.daily.partition.opaque;

import org.apache.storm.coordination.BatchOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseTransactionalBolt;
import org.apache.storm.transactional.ICommitter;
import org.apache.storm.transactional.TransactionAttempt;
import org.apache.storm.tuple.Tuple;

import java.math.BigInteger;
import java.util.HashMap;
import java.util.Map;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 20:22
 * @Email elonyong@163.com
 */
public class MyOpaqueCommitter extends BaseTransactionalBolt implements ICommitter {
    private static final long serialVersionUID = -1502810353308340730L;
    public static Map<String, DbValue> dbMap = new HashMap<>();
    Map<String, Integer> countMap = new HashMap<>();


    TransactionAttempt id;
    BatchOutputCollector batchOutputCollector;
    String today;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, BatchOutputCollector batchOutputCollector, TransactionAttempt transactionAttempt) {
        this.batchOutputCollector = batchOutputCollector;
        this.id = transactionAttempt;
    }

    @Override
    public void execute(Tuple tuple) {
        today = tuple.getString(1);
        Integer count = tuple.getInteger(2);
        id = (TransactionAttempt) tuple.getValue(0);
        if (today != null && count != null) {
            Integer batchCount = countMap.get(today);
            if (batchCount == null) {
                batchCount = 0;
            }
            batchCount += count;
            countMap.put(today, batchCount);
        }

    }

    @Override
    public void finishBatch() {
        if (countMap.size() > 0) {
            DbValue dbValue = dbMap.get(today);
            DbValue newValue;

            if (dbValue == null || !dbValue.txId.equals(id.getTransactionId())) {
                //更新数据库
                newValue = new DbValue();
                newValue.txId = id.getTransactionId();
                newValue.dateStr = today;
                if (dbValue == null) {
                    newValue.count = countMap.get(today);
                    newValue.pre_count = 0;
                } else {
                    newValue.pre_count = dbValue.count;
                    newValue.count = dbValue.count + countMap.get(today);
                }
                dbMap.put(today, newValue);
            } else {
                newValue = dbValue;
            }
            System.out.println("total============================:" + dbMap.get(today).count);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    static class DbValue {
        BigInteger txId;
        int count = 0;
        String dateStr;
        int pre_count;
    }
}
